package com.vip.iot.protocol.mqtt;

import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;

/**
 * MQTT实例实现
 *
 * @author wgb
 * @date 2020/6/17 18:16
 */
@Slf4j
public class MqttInstanceImpl implements IMqttInstance, IMqttMessageListener {
    /**
     * MQTT实例ID
     */
    private String clientId;
    /**
     * MQTT实例
     */
    private MqttClient mqttClient;
    /**
     * 消息订阅实例集合
     */
    private Map<String, OnMessageListener> subscribeInstanceMap;

    public MqttInstanceImpl(String clientId) {
        this.clientId = clientId;
    }

    public MqttInstanceImpl() {
        this(UUID.randomUUID().toString());
    }

    @Override
    public void init(String url, String username, String password) {
        // 创建消息订阅实例集合
        subscribeInstanceMap = new HashMap<>();
        // 持久化存储
        MemoryPersistence persistence = new MemoryPersistence();
        try {
            mqttClient = new MqttClient(url, this.clientId, persistence);
            MqttConnectOptions options = new MqttConnectOptions();
            options.setCleanSession(true);
            if (!"".equals(username)) {
                options.setUserName(username);
            }
            if (!"".equals(password)) {
                options.setPassword(password.toCharArray());
            }
            mqttClient.connect(options);
        } catch (MqttException e) {
            try {
                log.error("初始化MQTT实例出错：", e);
                Thread.sleep(5000);
                log.error("连接MQTT服务器失败，正在重连……");
                init(url, username, password);
            } catch (InterruptedException e1) {
                e1.printStackTrace();
            }
        }
    }

    @Override
    public void publish(String topic, String message, int qos) throws MqttException {
        MqttMessage mqttMessage = new MqttMessage(message.getBytes());
        mqttMessage.setQos(qos);
        mqttClient.publish(topic, mqttMessage);
    }

    @Override
    public void subscribe(String topic, OnMessageListener listener) throws MqttException {
        try {
            mqttClient.subscribe(topic, this);
            subscribeInstanceMap.put(topic, listener);
        } catch (MqttException e) {
            subscribeInstanceMap.remove(topic);
            throw e;
        }
    }

    @Override
    public void unsubscribe(String topic) throws MqttException {
        mqttClient.unsubscribe(topic);
        subscribeInstanceMap.remove(topic);
    }

    /**
     * 关闭MQTT实例连接，出现异常则抛出IO异常，释放MqttClient下的所有方法，禁止访问
     *
     * @throws IOException
     */
    @Override
    public void close() throws IOException {
        try{
            mqttClient.disconnect();
        }catch (MqttException e){
            throw new IOException();
        }
    }

    /**
     * 回调，收到的消息的处理
     * 因为是函数式接口，所以具体功能在实例中调用时可自定义
     *
     * @param topic 主题
     * @param mqttMessage 消息体
     */
    @Override
    public void messageArrived(String topic, MqttMessage mqttMessage) {
        // 获取监听器
        OnMessageListener listener = subscribeInstanceMap.get(topic);
        if(listener != null){
            listener.handleMessage(topic, mqttMessage.toString());
        }
    }
}
